package com.cnpc.bds.sdk.sink.kafka.executor;

import com.cnpc.bds.sdk.sink.kafka.meta.ConsumerCfg;

import java.util.List;

/**
 * 上游Kafka-消费者任务实现.
 */
public class DefaultUpstreamConsumer extends UpstreamConsumer {

    /**
     * 消息广播器.
     */
    private final EventBroadcaster downBroadcaster;

    public DefaultUpstreamConsumer(String name, ConsumerCfg consumerCfg) {
        super(name, consumerCfg);
        downBroadcaster = new EventBroadcaster(consumerCfg.getTopicMeta().getId() + "_down-handle-event-loop");
    }

    public DefaultUpstreamConsumer(ConsumerCfg consumerCfg) {
        super(consumerCfg);
        downBroadcaster = new EventBroadcaster(consumerCfg.getTopicMeta().getId() + "_down-handle-event-loop");
    }

    public static UpstreamConsumer buildAndExec(ConsumerCfg consumerCfg) {
        DefaultUpstreamConsumer upstreamConsumer = new DefaultUpstreamConsumer("task-default-upstream-consumer-0", consumerCfg);
        upstreamConsumer.start();
        return upstreamConsumer;
    }

    @Override
    public void consume(List<Object> msgList) {
        //向下委托广播.
        for (Object msg : msgList) {
            downBroadcaster.broadcast(msg);
        }
    }

    /**
     * 下游消息处理钩子.
     *
     * @return EventBroadcaster 默认消息广播器.
     */
    @Override
    public EventBroadcaster downStreamHook() {
        return downBroadcaster;
    }

}